import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;
import static org.apache.flink.table.api.Expressions.$;



public class Offset_Fetch {

    public static void main(String[] args) throws Exception {



        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

        DataSet<Order> ds = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4),
                new Order(1L, "diaper", 5),
                new Order(1L, "diaper", 6),
                new Order(1L, "diaper", 7),
                new Order(1L, "diaper", 8),
                new Order(1L, "diaper", 9),
                new Order(1L, "diaper", 10),
                new Order(1L, "diaper", 11),
                new Order(1L, "diaper", 12),
                new Order(1L, "diaper", 13),
                new Order(1L, "diaper", 14),
                new Order(1L, "diaper", 15)

        ));


        Table in = tEnv.fromDataSet(ds, "user,product,amount");

// returns the first 5 records from the sorted result
        Table result1 = in.orderBy($("amount").asc()).fetch(5);

// skips the first 3 records and returns all following records from the sorted result
        Table result2 = in.orderBy($("amount").asc()).offset(3);

// skips the first 10 records and returns the next 5 records from the sorted result
        Table result3 = in.orderBy($("amount").asc()).offset(10).fetch(5);


        tEnv.toDataSet(result3, Row.class).print();



    }
}
